package cn.me.alphamq.broker.event.spi.listener.broker;

import cn.me.alphamq.broker.cache.Container;
import cn.me.alphamq.broker.constant.BrokerConstant;
import cn.me.alphamq.broker.entity.RetryConsumeMsg;
import cn.me.alphamq.broker.event.model.broker.RetryConsumeMsgEvent;
import cn.me.alphamq.broker.model.timewheel.DelayMessageModel;
import cn.me.alphamq.broker.rebalance.ConsumerInstance;
import cn.me.alphamq.broker.rebalance.ConsumerInstanceManager;
import cn.me.alphamq.broker.store.TimeWheelModelManager;
import cn.me.alphamq.common.entity.Message;
import cn.me.alphamq.common.entity.req.RetryConsumeMsgReq;
import cn.me.alphamq.common.entity.resp.RetryConsumeMsgResp;
import cn.me.alphamq.common.enumeration.AckStatusEnum;
import cn.me.alphamq.common.enumeration.EventCodeEnum;
import cn.me.alphamq.common.event.Listener;
import cn.me.alphamq.common.msg.TcpMsg;
import cn.me.alphamq.common.util.DelayTimeUtil;
import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;

import static cn.me.alphamq.broker.constant.BrokerConstant.INTERVAL_TABLE;

/**
 * 重试消费消息的事件的监听器 TODO 这个消息还没有同步给从 broker
 *
 * @author f
 */
public class RetryConsumeMsgListener implements Listener<RetryConsumeMsgEvent> {

    private static final Logger log = LoggerFactory.getLogger(RetryConsumeMsgListener.class);

    @Override
    public void onReceive(RetryConsumeMsgEvent event) throws Exception {
        RetryConsumeMsgReq retryConsumeMsgReq = event.getRetryConsumeMsgReq();
        ChannelHandlerContext ctx = event.getCtx();
        InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();

        String msgId = retryConsumeMsgReq.getMsgId();
        String ip = inetSocketAddress.getHostString();
        int port = inetSocketAddress.getPort();
        String topic = retryConsumeMsgReq.getTopic();
        String consumerGroup = retryConsumeMsgReq.getConsumerGroup();
        int queueId = retryConsumeMsgReq.getQueueId();
        int maxRetryTimes = retryConsumeMsgReq.getMaxRetryTimes();

        // 校验消费者，如果失败，则重试失败
        RetryConsumeMsgResp retryConsumeMsgResp = new RetryConsumeMsgResp(msgId);
        if (!checkConsumer(topic, consumerGroup, ip, port)) {
            retryConsumeMsgResp.setRetryStatus(AckStatusEnum.FAIL.getCode());
            ctx.writeAndFlush(new TcpMsg(EventCodeEnum.RETRY_CONSUME_MSG_RESP,
                    JSON.toJSONBytes(retryConsumeMsgResp)));
            return;
        }

        retryConsumeMsgReq.getMessageList().forEach(delayMsg -> {
            // 跳过重试的消息
            Container.getConsumeQueueConsumeHandler().ack(topic, consumerGroup, queueId);

            // 构建重试消息，将其写入指定队列（重试/死信队列），并且将重试消息放到时间轮中
            try {
                int msgOffset = delayMsg.getMsgOffset();
                int msgLength = delayMsg.getMsgLength();
                int curRetryTimes = delayMsg.getCurRetryTimes() + 1;

                // 构建重试消息
                RetryConsumeMsg retryConsumeMsg = new RetryConsumeMsg(
                        topic, consumerGroup, msgOffset, msgLength, curRetryTimes
                );
                Message message = new Message();
                if (curRetryTimes > maxRetryTimes) {
                    // 如果超过重试次数上限，写入 死信队列
                    message.setTopic(BrokerConstant.DEAD_TOPIC_NAME);
                } else {
                    // 否则写入 重试队列
                    message.setTopic(BrokerConstant.RETRY_TOPIC_NAME);
                    long executeTime = DelayTimeUtil.getExecuteTime(INTERVAL_TABLE.get(curRetryTimes));
                    retryConsumeMsg.setNextRetryTime(executeTime);
                    // 并且将重试消息放到时间轮中 TODO 严格来讲，放到时间轮中的东西也应该持久化到文件中
                    TimeWheelModelManager.add(DelayMessageModel.newRetryMsgModel(retryConsumeMsg, executeTime));
                }
                message.setBody(JSON.toJSONBytes(retryConsumeMsg));

                // 将消息写入对应的队列
                Container.getCommitLogAppendHandler().appendMsg(message);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            }
        });

        // 返回响应
        retryConsumeMsgResp.setRetryStatus(AckStatusEnum.SUCCESS.getCode());
        ctx.writeAndFlush(new TcpMsg(EventCodeEnum.RETRY_CONSUME_MSG_RESP,
                JSON.toJSONBytes(retryConsumeMsgResp)));
    }

    // 校验消费者
    private boolean checkConsumer(String topic, String consumeGroup, String instanceIp, int instancePort) {
        // 如果该 topic 不存在，则重试失败
        if (Container.fetchCommitLogTopicModelMap().get(topic) == null) {
            log.error("topic [{}] does not exist", topic);
            return false;
        }

        // 如果该 topic 没有消费者实例，则重试失败
        Map<String, List<ConsumerInstance>> topicCIMap = Container.getConsumerInstanceMap().get(topic);
        if (topicCIMap == null || topicCIMap.isEmpty()) {
            log.error("topic [{}] has no consumer", topic);
            return false;
        }

        // 如果该 消费者组 没有消费者实例，则重试失败
        List<ConsumerInstance> consumerGroupCIList = topicCIMap.get(consumeGroup);
        if (consumerGroupCIList == null || consumerGroupCIList.isEmpty()) {
            log.error("consumeGroup{}] has no consumer", consumeGroup);
            return false;
        }

        // 如果该消费者实例与传来的信息不匹配，则重试失败
        String instanceId = ConsumerInstanceManager.getInstanceId(instanceIp, instancePort);
        if (consumerGroupCIList.stream().noneMatch(ci -> instanceId.equals(ConsumerInstanceManager.getInstanceId(ci)))) {
            log.error("consumeGroup{}] has no consumer", consumeGroup);
            return false;
        }

        return true;
    }
}
